
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

import java.util.Properties


object Question01 {
  def main(args: Array[String]): Unit = {
    // 创建 SparkSession 对象
    val spark =SparkUtils.getSparkSession()
    //设置日志级别
    spark.sparkContext.setLogLevel("WARN")

    // 读取用户信息数据集
    // csv : 文本文件的格式，分隔符规定是","
    val users: DataFrame = spark.read
      .option("header", "false") // 有没有字段名称
      .option("inferSchema", "true") //自动推断每一列的数据类型
      .csv(SparkUtils.usersPath)
      .toDF("uid", "username", "uage", "regtime")  // 设置列名

//    val rdd: RDD[String] = users.rdd.map(_.toString())
//    rdd.foreach(println)
    users.printSchema()

    // 读取充电桩使用记录数据集
    val charges = spark.read
      .option("header", "false")
      .option("inferSchema", "true")
      .csv(SparkUtils.chargingPath)
      .toDF("max", "min", "name")

    // 创建临时视图(临时表)
    users.createOrReplaceTempView("users")
    charges.createOrReplaceTempView("charges")

    //使用 Spark SQL 语句查询符合条件的用户: 平均消费超过40的用户和消费金额 - 用户名称 - 消费金额
    //SQL使用join...on...
    // GROUP BY users.uid,username    如果uid，username要显示，则username 不能省略
    //分组的sql语句 要显示的列：要么是分组的列，要么是聚合函数
//    val result = spark.sql(
//      """
//        |SELECT users.uid,username, SUM(money) AS total_charge_amount
//        |FROM charges JOIN users ON charges.uid = users.uid
//        |WHERE money > 0 AND username IS NOT NULL
//        |GROUP BY users.uid,username
//        |HAVING AVG(money) > 40
//        |order by total_charge_amount desc
//      """.stripMargin)

    //SQL另一种写法：使用 where
    var result1 = spark.sql(
//      """
//        |SELECT users.uid,username, SUM(money) AS total_charge_amount
//        |FROM charges,users where charges.uid = users.uid and  money > 0 AND username IS NOT NULL
//        |GROUP BY users.uid,username
//        |HAVING AVG(money) > 40
//        |order by total_charge_amount desc limit 10
//        |""".stripMargin)

      """
       |SELECT *
       |FROM charges limit 33

       |""".stripMargin)
    // 将结果保存为 CSV 文件
//

    //将结果保存到mysql表中
    val prop = new Properties()
    prop.setProperty("user","root")
    prop.setProperty("password","123456")
    result1.write.mode("append").jdbc(SparkUtils.url,"tianqi5" ,prop)
    spark.stop()
  }
}
